package com.atguigu.day5

import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.triggers.{Trigger, TriggerResult}
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

object TrigerExample {
  def main(args: Array[String]): Unit = {


    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    //    设置时间语义为事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    //    系统默认200毫秒插入一次水位线    这里设为1000毫秒

    //
    val stream = env.socketTextStream("192.168.35.102", 9999, '\n')

    val str = stream
      .map(line => {
        val arr = line.split(" ")
        (arr(0), arr(1).toLong)
      })
      .assignAscendingTimestamps(_._2)
      .keyBy(_._1)
      .timeWindow(Time.seconds(10))
      .trigger(new OnesecondIntervalTrigger)
      .process(new WindowContext)
      .print()
    env.execute()
  }

  /* a 1234
  a 2234
  a 12234*/
  class OnesecondIntervalTrigger extends Trigger[(String, Long), TimeWindow] {
    //每条数据调用一次
    override def onElement(element: (String, Long), timestamp: Long, window: TimeWindow, ctx: Trigger.TriggerContext): TriggerResult = {
      //默认值为false
      val firsrSeen = ctx.getPartitionedState(
        new ValueStateDescriptor[Boolean]("first-seen", Types.of[Boolean])
      )
      //当地一条数据来的时候，！firstSeen.value()设为true
      //这里的定时器指得是 ：onEventTime函数
      if (!firsrSeen.value()) {
        println("当前水位线是" + ctx.getCurrentWatermark)
        //如果当前水位线是1234，那么t = 1234 +(1000 - 1234 % 1000)=2000
        val t = ctx.getCurrentWatermark + (1000 - (ctx.getCurrentWatermark % 1000))
        //在第一条数据的时间戳之后的整数秒注册一个定时器
        ctx.registerEventTimeTimer(t)
        //在窗口结束事件注册一个定时器
        ctx.registerEventTimeTimer(window.getEnd)
        firsrSeen.update(true)

      }
      TriggerResult.CONTINUE

    }

    override def onProcessingTime(time: Long, window: TimeWindow, ctx: Trigger.TriggerContext): TriggerResult = {
      TriggerResult.CONTINUE
    }

    //定时器函数
    //当水位线到达时触发
    override def onEventTime(time: Long, window: TimeWindow, ctx: Trigger.TriggerContext): TriggerResult = {
      //在onElement函数中，我们注册过窗口时间的定时器
      //窗口闭合的默认操作会被overread
      if (time == window.getEnd) {
        //窗口闭合时，触发计算并清空窗口
        TriggerResult.FIRE_AND_PURGE
      } else {
        val t = ctx.getCurrentWatermark + (1000 - (ctx.getCurrentWatermark % 1000))
        if (time < window.getEnd) {
          ctx.registerEventTimeTimer(t)
        }
        //触发窗口计算
        TriggerResult.FIRE
      }
    }

    override def clear(window: TimeWindow, ctx: Trigger.TriggerContext): Unit = {
      val firsrSeen = ctx.getPartitionedState(
        new ValueStateDescriptor[Boolean]("first-seen", Types.of[Boolean])
      )
      firsrSeen.clear()
    }
  }

  class WindowContext extends ProcessWindowFunction[(String, Long), String, String, TimeWindow] {
    override def process(key: String, context: Context, elements: Iterable[(String, Long)], out: Collector[String]): Unit = {
      out.collect("窗口中有" + elements.size + "个元素.  窗口结束时间为" + context.window.getEnd)
    }
  }
}
